package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Demo22CheckPoint {
  def main(args: Array[String]): Unit = {

    /**
     * CheckPoint：检查点，为了实时Spark程序的容错性的保证
     * 可以将RDD的数据持久化到可靠的文件存储系统中
     */

    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName(this.getClass.getSimpleName.replace("$", ""))
    val sc = new SparkContext(conf)

    // 需要指定目录一般是可靠的文件存储系统目录，例如：HDFS
    sc.setCheckpointDir("spark/data/ck")

    val stuRDD: RDD[String] = sc.textFile("spark/data/stu/students.txt")

    val mapStuRDD: RDD[String] = stuRDD.map(line => {
      println("执行了map方法")
      line
    })

    // 对将要进行CK的RDD可以先进行cache
    mapStuRDD.cache()

    // 直接对RDD可以进行CheckPoint，但是需要提前设置好CheckPoint的路径
    mapStuRDD.checkpoint() // RDD在进行CK时，会单独启动一个计算任务用于将CK的数据保存起来

    // 统计班级人数
    mapStuRDD.map(line => (line.split(",")(4), 1)).reduceByKey(_ + _).foreach(println)

    // 统计性别人数
    mapStuRDD.map(line => (line.split(",")(3), 1)).reduceByKey(_ + _).foreach(println)

    // 统计年龄人数
    mapStuRDD.map(line => (line.split(",")(2), 1)).reduceByKey(_ + _).foreach(println)
    mapStuRDD.map(line => (line.split(",")(2), 1)).reduceByKey(_ + _).foreach(println)

    while (true) {

    }

  }

}
